package com.atguigu.userprofile.bitmap.app

import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.constant.ConstCode
import com.atguigu.userprofile.dao.TagDAO
import com.atguigu.userprofile.util.MyClickhouseUtil
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object TaskBitmapApp {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("task_bitmap_app").setMaster("local[*]")
    val sparkContext = new SparkContext(sparkConf)
    val taskDateOrigin: String = args(1)


    //  1)	读取标签定义表，用于获得标签名称和标签类型。
    val tagInfoList: List[TagInfo] = TagDAO.getTagInfoOnTaskList()

    //  2)	根据标签类型的不同要写入到四种不同的Bitmap表中。
    val tagStringList = new ListBuffer[TagInfo]
    val tagDateList = new ListBuffer[TagInfo]
    val tagDecimalList = new ListBuffer[TagInfo]
    val tagLongList = new ListBuffer[TagInfo]
    for (tagInfo <- tagInfoList) {
      if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_STRING) {
        tagStringList.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_DATE) {
        tagDateList.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_LONG) {
        tagLongList.append(tagInfo)
      } else if (tagInfo.tagValueType == ConstCode.TAG_VALUE_TYPE_DECIMAL) {
        tagDecimalList.append(tagInfo)
      }
    }

    insertBitmap(tagStringList.toList, "user_tag_value_string", taskDateOrigin)
    insertBitmap(tagLongList.toList, "user_tag_value_long", taskDateOrigin)
    insertBitmap(tagDecimalList.toList, "user_tag_value_decimal", taskDateOrigin)
    insertBitmap(tagDateList.toList, "user_tag_value_date", taskDateOrigin)


  }


  //insert into  $user_tag_value_string
  //SELECT
  //    tag_code_value.1 AS tag_code,
  //    tag_code_value.2 AS tag_value,
  //   groupBitmapState (uid) AS us
  //FROM
  //(
  //    SELECT
  //        arrayJoin([('AGEGROUP', agegroup), ('GENDER', gender), ('FAVOR', favor)]) AS tag_code_value,    //拼接 tagcodes
  //        uid
  //    FROM user_tag_merge_$taskDate
  //) AS tv
  //GROUP BY
  //    tag_code_value.1,
  //    tag_code_value.2
  def insertBitmap(tagCodeTypeList: List[TagInfo], tableName: String, taskDateOrigin: String): Unit = {
    if (tagCodeTypeList.size > 0) {
      val tagCodesSql: String = tagCodeTypeList.map(tagInfo => s"('${tagInfo.tagCode}', ${tagInfo.tagCode.toLowerCase()})").mkString(",")
      val sql =
        s"""
           | insert into  $tableName
           |  SELECT
           |      tag_code_value.1 AS tag_code,
           |     tag_code_value.2 AS tag_value,
           |     groupBitmapState (toUInt64(uid)) AS us,
           |     '$taskDateOrigin'
           |  FROM
           |  (
           |     SELECT
           |          arrayJoin([ $tagCodesSql]) AS tag_code_value,
           |          uid
           |      FROM user_tag_merge_${taskDateOrigin.replace("-", "")}
           |
         |  ) AS tv
           |  where tag_code_value.2<>''
           |  GROUP BY
           |      tag_code_value.1,
           |      tag_code_value.2
       """.stripMargin

      MyClickhouseUtil.executeSql(sql)
    }
  }

}
